package com.fjeld.middleware.activemq;

import com.fjeld.middleware.constants.PropertiesManager;
import com.fjeld.middleware.jaxb.V1Message;
import org.apache.log4j.Logger;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by IntelliJ IDEA.
 * User: fjeld
 * Date: 03/01/2012
 * Time: 00:22
 *
 * Publishes the messages in the provided queue
 */
public class QueuedMessagePublisher extends TopicPublisher implements Runnable
{
    private static final Logger LOGGER = Logger.getLogger(QueuedMessagePublisher.class);
    private BlockingQueue<V1Message> messageQueue;

    public QueuedMessagePublisher(String topicString, String publisherName)
    {
        super(topicString, publisherName);
        LOGGER.info(publisherName);

        String queueSizeString = PropertiesManager.getProperties().getProperty(
                "QueuedMessageListener.QueueSize", "1000");
        Integer queueSize = Integer.parseInt(queueSizeString);
        LOGGER.info("QueueSize: " + queueSize);

        messageQueue = new LinkedBlockingQueue<V1Message>(queueSize);
    }

    public void startConnection()
    {
        super.startConnection();
        new Thread(this).start();
    }

    public void run()
    {
        try
        {
            while (true)
            {
                LOGGER.info("Waiting for message to publish...");
                V1Message v1Message = messageQueue.take();
                v1Message.setSendingTime(System.currentTimeMillis());
                LOGGER.info("Publishing message.");
                publish(v1Message);
                LOGGER.info("Published.");
            }
        } catch (Exception e)
        {
            LOGGER.error(e);
        }
    }

    public BlockingQueue<V1Message> getMessageQueue()
    {
        return messageQueue;
    }
}
